package db.function

import com.cw.realtime.common.bean.TableProcessDwd
import com.cw.realtime.common.util.JdbcUtil
import io.circe.JsonObject
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.util.Collector

import scala.collection.mutable

class BcProcFunction4BaseDb(bcDesc: MapStateDescriptor[String, TableProcessDwd]) extends BroadcastProcessFunction[JsonObject, TableProcessDwd, (JsonObject, TableProcessDwd)]{

  private type ResType = (JsonObject, TableProcessDwd)
  private type FuncType = BroadcastProcessFunction[JsonObject, TableProcessDwd, ResType]
  val map = mutable.Map.empty[String, TableProcessDwd]

  override def open(parameters: Configuration): Unit = {
    for {
      conn <- JdbcUtil.getMysqlConnection
      tpd <- JdbcUtil.queryList[TableProcessDwd](conn, "select * from gmall_config.table_process_dwd")
      key = tpd.sourceTable + ":" + tpd.sourceType
    } {
      map += key -> tpd
    }
  }

  override def processBroadcastElement(value: TableProcessDwd, ctx: FuncType#Context, out: Collector[ResType]): Unit = {
    val key = value.sourceTable + ":" + value.sourceType
    val state = ctx.getBroadcastState(bcDesc)
    if ("d" == value.op) {
      state.remove(key)
      map -= key
    } else
      state.put(key, value)
  }

  override def processElement(value: JsonObject, ctx: FuncType#ReadOnlyContext, out: Collector[ResType]): Unit = {
    for {
      recordType <- value("type")
      recordType <- recordType.asString
      table <- value("table")
      table <- table.asString
      state <- Option(ctx.getBroadcastState(bcDesc))
      key = table + ":" + recordType
    } {
      Option(state.get(key)) match {
        case Some(tpd) => out.collect(value -> tpd)
        case None => map.get(key) match {
          case Some(tpd) => out.collect(value -> tpd)
          case None =>
        }
      }
    }
  }

}
